package com.it.table.time_window;

import com.it.pojo.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * 使用窗口的方式实现Top N
 *
 * @author code1997
 */
public class TopNByWindow {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置水位线
        SingleOutputStreamOperator<Event> eventSource = executionEnvironment.fromElements(
                        new Event("tom", "./index", 1000L),
                        new Event("tom", "./product?id=10", 2000L),
                        new Event("lily", "./index", 3000L),
                        new Event("tom", "./order?id=10", 40000L),
                        new Event("ailis", "./index", 5000L),
                        new Event("tom", "./pay?id=10", 16000L),
                        new Event("tom", "./pay?id=1000", 20000L))
                .assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                            @Override
                            public long extractTimestamp(Event element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        }));
        //创建表环境
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(executionEnvironment);
        //流转换为表
        Table sourceTable = streamTableEnvironment.fromDataStream(eventSource, $("user"), $("url"), $("timestamp"), $("timestamp").rowtime().as("ts"));
        //注册为临时表
        streamTableEnvironment.createTemporaryView("event_source", sourceTable);
        //定义TVF
        String subQuery = "SELECT window_start, window_end, `user`, COUNT(url) as cnt " +
                "FROM TABLE ( " +
                "TUMBLE (TABLE event_source, DESCRIPTOR(ts), INTERVAL '10' SECOND))" +
                "GROUP BY window_start,window_end,`user`";
        String midQuery = "SELECT sub_view.window_start, sub_view.window_end, sub_view.`user`, sub_view.cnt ," +
                "row_number() over (partition by window_start,window_end order by cnt desc) as rown " +
                "FROM (" +
                subQuery +
                ") AS sub_view";
        String topQuery = "SELECT * FROM (" +
                midQuery +
                ") AS top where top.rown <= 2";

        Table result = streamTableEnvironment.sqlQuery(topQuery);
        streamTableEnvironment.toDataStream(result).print();
        executionEnvironment.execute();
    }
}
